package udf.matching;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.Iterator;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

public class JoinUsingBagOfBag extends EvalFunc<DataBag> {

	private boolean initialized = false;
	private int bag_entry_size = -1;
	private BagFactory bagFactory = BagFactory.getInstance();

	// static ArrayList<String> mTables = new ArrayList<String>();

	// static HashSet<String> seeds = new HashSet<String>();
	private Hashtable<String, String[]> seeds = new Hashtable<String, String[]>();

	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";

	public void init(Tuple tuple) throws IOException {

		DataBag bag = (DataBag) tuple.get(1);  //bag of bags

		String tag_pos_ = (String) tuple.get(2);
		int tag_pos = Integer.valueOf(tag_pos_);

		Iterator<Tuple> iterator = bag.iterator();

		while (iterator.hasNext()) {
			
			
			
			Tuple tup = iterator.next();
			System.out.println("Tuple: "+ tup.size());
			System.out.println(tup);
			String temp[] = new String[tup.size()];
			for (int i = 0; i < tup.size(); i++) {
				temp[i] = (String) tup.get(i);
			}

			bag_entry_size = temp.length;

			String tag = (String) tup.get(tag_pos);
			// System.out.println("Adding: " + tag + "\t" + temp[0]);
			seeds.put(tag, temp);

		}

		initialized = true;
	}

	public DataBag exec(Tuple input) throws IOException {

		if (input.size() != 4) {
			return null;
		}

		else {

			if (!initialized)
				init(input);

			Tuple tags = (Tuple) input.get(0);
			String pos_ = (String) input.get(3);
			int pos = Integer.valueOf(pos_);
			ArrayList<Integer> found = new ArrayList<Integer>();
			ArrayList<String[]> lines = new ArrayList<String[]>();

		
			for (int i = 0; i < tags.size(); i++) {

				String tag = (String) tags.get(i);

				if (seeds.containsKey(tag)) {
					found.add(i);
					lines.add(seeds.get(tag));
				//	System.out.println("Matching " + tag);
				}

			}
			DataBag output = bagFactory.newDefaultBag();

			if (found.size() > 0) {
			//	System.out.println("Found size: "+ found);
				for (int k = 0; k < found.size(); k++) {
					
					int index = found.get(k);
					
					Tuple t = TupleFactory.getInstance().newTuple(tags.size() + bag_entry_size);
					String temporal = (String) tags.get(pos);
					String target = (String) tags.get(index);

					for (int i = 0; i < tags.size(); i++) {
						if (i == pos) {
							t.set(i, target);

						} else if (i == index) {
							t.set(i, temporal);

						} else {
							t.set(i, tags.get(i));
						}
					}
					int j = 0;
					String line[] = lines.get(k);

					for (int i = tags.size(); i < t.size(); i++) {
						t.set(i, line[j]);
						j++;
					}
				output.add(t);
				}

			} else {

				return null;
			}

			return output;

		}

	}
}